package join;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.lib.MultipleInputs;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class ClickUserAuctionInnerJoinPv extends Configured implements Tool {

	private static final String tab_separator = "\t";
	private static final String pvAuctionSeparator1 = "\\+";
	private static final String pvAuctionSeparator2 = "\\|";
	private static HashMap<String, String> sortMap = null;

	/*****
	 * (PV.4用户ID != null) && (!PV4.用户ID.equals("")) && <br/>
	 * (ClickUserAuction.4Userid != "-") && <br/>
	 * PV.5query != null && ClickUserAuction.2Query != null && <br/>
	 * !PV.5query.equals("") && !ClickUserAuction.2Query.equals("")<br/>
	 *****/
	/*****
	 * JOIN条件: <br/>
	 * for(Record PV in PV Log, Record ClickUserAuction in ClickUserAuction Log)
	 * {<br/>
	 * if ((|PV.3请求时间 - ClickUserAuction.9点击时间| < 2s) && <br/>
	 * (PV.9商品ID == ClickUserAuction.0商品数字ID) && <br/>
	 * (PV.4用户ID == ClickUserAuction.5Userid) && <br/>
	 * (PV.5query == ClickUserAuction.3Query) && <br/>
	 * (PV.6类目 == ClickUserAuction.11搜索类目) && <br/>
	 * (PV.7属性串 == ClickUserAuction.26搜索属性))<br/>
	 * (PV.8搜索排序方式 == ClickUserAuction.29排序) && <br/>
	 * then Join(PV, ClickUserAuction) <br/>
	 * }
	 *****/
	/*****
	 * PV格式: <br/>
	 * 0 <br/>
	 * 1 Acookie的cookieid <br/>
	 * 2 Tcookie的cookieid <br/>
	 * 3 请求时间 <br/>
	 * 4 用户id（32位ID） (新版为空)<br/>
	 * 5 query <br/>
	 * 6 类目【直达后的类目】 <br/>
	 * 7 属性串 <br/>
	 * 8 搜索的排序方式，格式如：_ps=coefp&ss=ends <br/>
	 * 9 展现的商品id<br/>
	 *****/
	// public static class PvMap extends MapReduceBase implements
	// Mapper<LongWritable, Text, Text, Text> {
	// private Text k = new Text(), v = new Text();
	//
	// public void map(LongWritable key, Text value,
	// OutputCollector<Text, Text> output, Reporter reporter)
	// throws IOException {
	// String line = value.toString();
	// String[] pv = line.split(tab_separator);
	// if (pv.length == 10 && isValid(pv[9]) && isValid(pv[4])
	// && isValid(pv[5])) {
	// String[] idss = pv[9].split(pvAuctionSeparator1);
	// for (int i = 0; i < idss.length; i++) {
	// if (isValid(idss[i])) {
	// String[] ids = idss[i].split(pvAuctionSeparator2);
	// sortFilterSetUp();
	// for (String itemNumID : ids) {
	// if (isValid(itemNumID)) {
	// System.out.println(3);
	// k.set(itemNumID + tab_separator
	// + pv[4]
	// + tab_separator
	// + pv[5] + tab_separator
	// + pv[6] + tab_separator + pv[7]
	// + tab_separator + sortFilt(pv[8]));
	// v.set(line.substring(0, line
	// .lastIndexOf(tab_separator)));
	// System.out
	// .println("PvMap key: " + k.toString());
	// System.out.println("PvMap value: "
	// + v.toString());
	// output.collect(k, v);
	// }
	// }
	// }
	// }
	// }
	// }
	// }
	public static class PvMap extends MapReduceBase implements
			Mapper<LongWritable, Text, Text, Text> {
		private Text k = new Text(), v = new Text();

		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			String line = value.toString();
			String[] pv = line.split(tab_separator);
			if (pv.length == 10 && isValid(pv[9]) && isValid(pv[4])
					&& isValid(pv[5])) {
				String[] idss = pv[9].split(pvAuctionSeparator1);
				for (int i = 0; i < idss.length; i++) {
					if (isValid(idss[i])) {
						String[] ids = idss[i].split(pvAuctionSeparator2);
						System.out.println("idss[i]" + idss[i]);
						sortFilterSetUp();
						for (String itemNumID : ids) {
							if (isValid(itemNumID)) {
								k.set(itemNumID + tab_separator + pv[4]
										+ tab_separator + pv[5] + tab_separator
										+ pv[6] + tab_separator + pv[7]
										+ tab_separator + sortFilt(pv[8]));
								v.set(line.substring(0, line
										.lastIndexOf(tab_separator)));
								output.collect(k, v);
							}
						}
					}
				}
			}
		}
	}

	/*****
	 * ClickUserAuction格式: <br/>
	 * 0 商品数字ID <br/>
	 * 1 32位UserID <br/>
	 * 2 Type（c1：结果页） <br/>
	 * 3 Query <br/>
	 * 4 bucket号（占用Query_type，如果对应不到bucket号，则为空，其它为0-19的值） <br/>
	 * 5 User id <br/>
	 * 6 用户分类 <br/>
	 * 7 Machine id <br/>
	 * 8 Session ID <br/>
	 * 9 点击时间 <br/>
	 * 10 搜索的来源pid <br/>
	 * 11 搜索类目 <br/>
	 * 12 上一步点击模块ID（智能导航或相关搜索） <br/>
	 * 13 排序 <br/>
	 * 14 列表模式（大图、小图） <br/>
	 * 15 消保筛选 <br/>
	 * 16 所在地筛选 <br/>
	 * 17 每页商品数 <br/>
	 * 18 第几页 <br/>
	 * 19 被点击的商品在商品列表的第几个位置 <br/>
	 * 20 点击商品的auctionId(32位ID) <br/>
	 * 21 商品所属类目，用auctionId关联 <br/>
	 * 22 商品所属属性，用auctionId关联 (目前Hadoop上没有该字段，需要导入) <br/>
	 * 23 商品所属的spuid <br/>
	 * 24 商品所属卖家ID <br/>
	 * 25 B/C商品标识 <br/>
	 * 26 搜索属性 <br/>
	 * 27 Tab类型 (0:未知，1：所有宝贝，2：人气，3：淘宝商城，4：二手) <br/>
	 * 28 是否豆腐块商品（0:否，1：是） <br/>
	 * 29 统一排序方式 <br/>
	 *****/
	/*****
	 * JOIN条件: <br/>
	 * for(Record PV in PV Log, Record ClickUserAuction in ClickUserAuction Log)
	 * {<br/>
	 * if ((|PV.3请求时间 - ClickUserAuction.9点击时间| < 2s) && <br/>
	 * (PV.9商品ID == ClickUserAuction.0商品数字ID) && <br/>
	 * (PV.4用户ID == ClickUserAuction.1Userid) && <br/>
	 * (PV.5query == ClickUserAuction.3Query) && <br/>
	 * (PV.6类目 == ClickUserAuction.11搜索类目) && <br/>
	 * (PV.7属性串 == ClickUserAuction.26搜索属性))<br/>
	 * (PV.8搜索排序方式 == ClickUserAuction.29排序) && <br/>
	 * then Join(PV, ClickUserAuction) <br/>
	 * }
	 *****/
	/***** KEY: 99405571 98683517 milestone 0 sale_desc *****/
	// public static class ClickUserAuctionMap extends MapReduceBase implements
	// Mapper<LongWritable, Text, Text, Text> {
	// private Text k = new Text(), v = new Text();
	//
	// public void map(LongWritable key, Text value,
	// OutputCollector<Text, Text> output, Reporter reporter)
	// throws IOException {
	// String line = value.toString();
	// String[] click = line.split(tab_separator, 30);
	// if (click.length == 30 && isValid(click[0])
	// && !click[5].equals("-")) {
	// k.set(click[0] + tab_separator + click[1] + tab_separator
	// + click[3] + tab_separator + click[11] + tab_separator
	// + click[26] + tab_separator + click[29]);
	// v.set(line);
	// System.out.println("ClickUserAuctionMap key: " + k.toString());
	// System.out
	// .println("ClickUserAuctionMap value: " + v.toString());
	// output.collect(k, v);
	// }
	// }
	// }
	public static class ClickUserAuctionMap extends MapReduceBase implements
			Mapper<LongWritable, Text, Text, Text> {
		private Text k = new Text(), v = new Text();

		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			String line = value.toString();
			String[] click = line.split(tab_separator, 30);
			if (click.length == 30 && isValid(click[0])
					&& !click[5].equals("-")) {
				k.set(click[0] + tab_separator + click[1] + tab_separator
						+ click[3] + tab_separator + click[11] + tab_separator
						+ click[26] + tab_separator + click[29]);
				v.set(line);
				output.collect(k, v);
			}
		}
	}

	/*****
	 * PvClick格式: <br/>
	 ** 0 商品数字ID <br/>
	 * 1 32位UserID <br/>
	 * 2 Type（c1：结果页） <br/>
	 ** 3 Query <br/>
	 * 4 bucket号（占用Query_type，如果对应不到bucket号，则为空，其它为0-19的值） <br/>
	 ** 5 User id <br/>
	 * 6 用户分类 <br/>
	 * 7 Machine id <br/>
	 * 8 Session ID <br/>
	 * 9 点击时间 <br/>
	 * 10 搜索的来源pid <br/>
	 ** 11 搜索类目 <br/>
	 * 12 上一步点击模块ID（智能导航或相关搜索） <br/>
	 * 13 排序 <br/>
	 * 14 列表模式（大图、小图） <br/>
	 * 15 消保筛选 <br/>
	 * 16 所在地筛选 <br/>
	 * 17 每页商品数 <br/>
	 * 18 第几页 <br/>
	 * 19 被点击的商品在商品列表的第几个位置 <br/>
	 * 20 点击商品的auctionId(32位ID) <br/>
	 * 21 商品所属类目，用auctionId关联 <br/>
	 * 22 商品所属属性，用auctionId关联 (目前Hadoop上没有该字段，需要导入) <br/>
	 * 23 商品所属的spuid <br/>
	 * 24 商品所属卖家ID <br/>
	 * 25 B/C商品标识 <br/>
	 ** 26 搜索属性 <br/>
	 * 27 Tab类型 (0:未知，1：所有宝贝，2：人气，3：淘宝商城，4：二手) <br/>
	 * 28 是否豆腐块商品（0:否，1：是） <br/>
	 ** 29 统一排序方式 <br/>
	 ** 30 Acookie的cookieid <br/>
	 ** 31 Tcookie的cookieid <br/>
	 ** 32 请求时间 <br/>
	 *****/
	// public static class Reduce extends MapReduceBase implements
	// Reducer<Text, Text, Text, Text> {
	// private Text CLICK = new Text("1");
	// private Text NOTCLICK = new Text("-1");
	// private Text v = new Text();
	//
	// public void reduce(Text key, Iterator<Text> values,
	// OutputCollector<Text, Text> output, Reporter reporter)
	// throws IOException {
	// ArrayList<String> pvContainer = new ArrayList<String>();
	// ArrayList<String> clickUserAuctionContainer = new ArrayList<String>();
	// while (values.hasNext()) {
	// String result = values.next().toString();
	// if (result.split(tab_separator).length == 9) {
	// pvContainer.add(result);
	// } else if (result.split(tab_separator).length == 30) {
	// clickUserAuctionContainer.add(result);
	// }
	// }
	// if (pvContainer.isEmpty() || clickUserAuctionContainer.isEmpty())
	// System.out.println("***********Empty***********");
	// if (!pvContainer.isEmpty()) {
	// sortFilterSetUp();
	// for (String pv : pvContainer) {
	// String[] pvArr = pv.split(tab_separator, 9);
	// System.out.println("pvArr: " + pvArr[0]);
	// for (String clickUserAuction : clickUserAuctionContainer) {
	// String[] clickUserAuctionArr = clickUserAuction.split(
	// tab_separator, 30);
	// System.out.println("clickUserAuctionArr: "
	// + clickUserAuctionArr[0]);
	// if ((Math.abs(Long.parseLong(pvArr[3])
	// - Long.parseLong(clickUserAuctionArr[9]))) < 5) {
	// v.set(clickUserAuction + tab_separator + pvArr[1]
	// + tab_separator + pvArr[2] + tab_separator
	// + pvArr[3]);
	// output.collect(CLICK, v);
	// }
	// break;
	// }
	// String keyStr = key.toString();
	// String itemID = keyStr.substring(0, keyStr
	// .indexOf(tab_separator));
	// String[] resultSet = new String[33];
	// for (int i = 0; i < resultSet.length; i++)
	// resultSet[i] = "";
	// resultSet[0] = itemID;
	// resultSet[3] = pvArr[5];
	// resultSet[5] = pvArr[4];
	// resultSet[11] = pvArr[6];
	// resultSet[26] = pvArr[7];
	// resultSet[29] = sortMap.get(pvArr[8]);
	// resultSet[30] = pvArr[1];
	// resultSet[31] = pvArr[2];
	// resultSet[32] = pvArr[3];
	// String res = new String("");
	// for (int i = 0; i < resultSet.length; i++) {
	// res += resultSet[i] + tab_separator;
	// }
	// v.set(res);
	// // output.collect(NOTCLICK, new Text(res));
	// }
	// }
	// }
	// }
	public static class Reduce extends MapReduceBase implements
			Reducer<Text, Text, Text, Text> {
		private Text CLICK = new Text("1");
		private Text NOTCLICK = new Text("-1");
		private Text v = new Text();

		public void reduce(Text key, Iterator<Text> values,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			ArrayList<String> pvContainer = new ArrayList<String>();
			ArrayList<String> clickUserAuctionContainer = new ArrayList<String>();
			while (values.hasNext()) {
				String result = values.next().toString();
				if (result.split(tab_separator).length == 9) {
					pvContainer.add(result);
				} else if (result.split(tab_separator).length == 30) {
					clickUserAuctionContainer.add(result);
				}
			}
			if (pvContainer.isEmpty() && !clickUserAuctionContainer.isEmpty()) {
				for (String click : clickUserAuctionContainer) {
					v
							.set(click + tab_separator + tab_separator
									+ tab_separator);
					output.collect(NOTCLICK, v);
				}
			}
			if (clickUserAuctionContainer.isEmpty() && !pvContainer.isEmpty()) {
				for (String pv : pvContainer) {
					String[] pvArr = pv.split(tab_separator);
					String keyStr = key.toString();
					String itemID = keyStr.substring(0, keyStr
							.indexOf(tab_separator));
					String[] resultSet = new String[33];
					for (int i = 0; i < resultSet.length; i++)
						resultSet[i] = "";
					resultSet[0] = itemID;
					resultSet[3] = pvArr[5];
					resultSet[5] = pvArr[4];
					resultSet[11] = pvArr[6];
					resultSet[26] = pvArr[7];
					resultSet[29] = sortMap.get(pvArr[8]);
					resultSet[30] = pvArr[1];
					resultSet[31] = pvArr[2];
					resultSet[32] = pvArr[3];
					String res = new String("");
					for (int i = 0; i < resultSet.length; i++) {
						res += resultSet[i] + tab_separator;
					}
					v.set(res);
					output.collect(NOTCLICK, new Text(res));
				}
			}
			if (!pvContainer.isEmpty() && !clickUserAuctionContainer.isEmpty()) {
				sortFilterSetUp();
				for (String pv : pvContainer) {
					String[] pvArr = pv.split(tab_separator, 9);
					for (String clickUserAuction : clickUserAuctionContainer) {
						String[] clickUserAuctionArr = clickUserAuction.split(
								tab_separator, 30);
						System.out.println(Long.parseLong(pvArr[3]));
						System.out.println(clickUserAuctionArr[9]);
						if ((Math.abs(Long.parseLong(pvArr[3])
								- Long.parseLong(clickUserAuctionArr[9]))) < 2) {
							v.set(clickUserAuction + tab_separator + pvArr[1]
									+ tab_separator + pvArr[2] + tab_separator
									+ pvArr[3]);
							output.collect(CLICK, v);
						} else {
							String keyStr = key.toString();
							String itemID = keyStr.substring(0, keyStr
									.indexOf(tab_separator));
							String[] resultSet = new String[33];
							for (int i = 0; i < resultSet.length; i++)
								resultSet[i] = "";
							resultSet[0] = itemID;
							resultSet[3] = pvArr[5];
							resultSet[5] = pvArr[4];
							resultSet[11] = pvArr[6];
							resultSet[26] = pvArr[7];
							resultSet[29] = sortMap.get(pvArr[8]);
							resultSet[30] = pvArr[1];
							resultSet[31] = pvArr[2];
							resultSet[32] = pvArr[3];
							String res = new String("");
							for (int i = 0; i < resultSet.length; i++) {
								res += resultSet[i] + tab_separator;
							}
							v.set(res);
							output.collect(NOTCLICK, new Text(res));
						}
					}

				}
			}
		}
	}

	public static boolean isValid(String src) {
		return (src != null && !src.equals("") && !src.equals("-"));
	}

	public static void sortFilterSetUp() {
		sortMap = new HashMap<String, String>();
		// bid,reserve_price,price_asc ps=reserve_price&ss=ends price_asc
		// 按照价格正序
		sortMap.put("bid", "price_asc");
		sortMap.put("reserve_price", "price_asc");
		sortMap.put("price_asc", "price_asc");
		sortMap.put("ps=reserve_price&ss=ends", "price_asc");
		// _bid,_reserve_price,price_desc _ps=reserve_price&ss=ends
		// price_desc 按照价格逆序
		sortMap.put("_bid", "price_desc");
		sortMap.put("_reserve_price", "price_desc");
		sortMap.put("price_desc", "price_desc");
		sortMap.put("_ps=reserve_price&ss=ends", "price_desc");
		// coefp ps=coefp&ss=ends coefp_desc 人气排序正序
		sortMap.put("coefp", "coefp_desc");
		sortMap.put("ps=coefp&ss=ends", "coefp_desc");
		// _coefp _ps=coefp&ss=ends coefp_asc 人气排序逆序
		sortMap.put("_coefp", "coefp_asc");
		sortMap.put("_ps=coefp&ss=ends", "coefp_asc");
		// _isnewest, isnewest _ps=isnewest&ss=ends, ps=isnewest&ss=ends
		// isnewest 最新上架产品优先
		sortMap.put("_isnewest", "isnewest");
		sortMap.put("isnewest", "isnewest");
		sortMap.put("_ps=isnewest&ss=ends", "isnewest");
		sortMap.put("ps=isnewest&ss=ends", "isnewest");
		// ratesum, credit_asc ps=ratesum&ss=ends ratesum_asc
		sortMap.put("ratesum", "ratesum_asc");
		sortMap.put("credit_asc", "ratesum_asc");
		sortMap.put("ps=ratesum&ss=ends", "ratesum_asc");
		// _ratesum，credit_desc _ps=ratesum&ss=ends ratesum_desc
		sortMap.put("_ratesum", "ratesum_desc");
		sortMap.put("credit_desc", "ratesum_desc");
		sortMap.put("_ps=ratesum&ss=ends", "ratesum_desc");
		// sale_asc ps=volume30day&ss=ends sale_asc
		sortMap.put("sale_asc", "sale_asc");
		sortMap.put("ps=volume30day&ss=ends", "sale_asc");
		// sale_desc _ps=volume30day&ss=ends sale_desc
		sortMap.put("sale_desc", "sale_desc");
		sortMap.put("_ps=volume30day&ss=ends", "sale_desc");
		// _ends, _all, _commend _ps=commend&ss=ends,
		// _ps=ends&ss=reserve_price, _ps=assure&_ss=commend all
		sortMap.put("_ends", "all");
		sortMap.put("_all", "all");
		sortMap.put("_commend", "all");
		sortMap.put("_ps=commend&ss=ends", "all");
		sortMap.put("_ps=ends&ss=reserve_price", "all");
		sortMap.put("_ps=assure&_ss=commend", "all");
		// end, all,commend ps=ends&ss=reserve_price,ps=commend&ss=ends,
		// ps=assure&_ss=commend all
		sortMap.put("end", "all");
		sortMap.put("all", "all");
		sortMap.put("commend", "all");
		sortMap.put("ps=ends&ss=reserve_price", "all");
		sortMap.put("ps=commend&ss=ends", "all");
		sortMap.put("ps=assure&_ss=commend", "all");
	}

	private static String sortFilt(String src) {
		return sortMap.containsKey(src) ? sortMap.get(src) : "other";
	}

	public int run(String[] args) throws Exception {
		String date = args[0];
		// ClickUserAuction
		String clickUserAuctionPathStr = "/group/tbsc-dev/xzwang/cyliu/join/intermediate/"
				+ date + "/clickUserAuction/";
		Path clickUserAuctionPath = new Path(clickUserAuctionPathStr);
		// Output
		String pvClickPathStr = "/group/tbsc-dev/xzwang/cyliu/join/output/"
				+ date + "/pvClick";
		Path pvClickPath = new Path(pvClickPathStr);
		JobConf conf = new JobConf(getConf(), ClickUserAuctionInnerJoinPv.class);
		conf.setJobName("PvClickJoin");
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);
		// Pv
		for (int i = 0; i < 3; i++)
			for (int j = 0; j < 10; j++) {
				int n = i * 10 + j;
				if (n >= 0 && n <= 23) {
					String hour = "";
					if (n < 10)
						hour = "0" + n;
					else
						hour = String.valueOf(n);
					MultipleInputs.addInputPath(conf, new Path(
							"/group/tbads/logdata/new_search_pv/" + date + "/"
									+ hour
									+ "/00/ttslave27.sds.cnz.alimama.com"),
							SequenceFileInputFormat.class, PvMap.class);
					MultipleInputs.addInputPath(conf, new Path(
							"/group/tbads/logdata/new_search_pv/" + date + "/"
									+ hour
									+ "/00/ttslave28.sds.cnz.alimama.com"),
							SequenceFileInputFormat.class, PvMap.class);
					MultipleInputs.addInputPath(conf, new Path(
							"/group/tbads/logdata/new_search_pv/" + date + "/"
									+ hour
									+ "/00/ttslave29.sds.cnz.alimama.com"),
							SequenceFileInputFormat.class, PvMap.class);
					MultipleInputs.addInputPath(conf, new Path(
							"/group/tbads/logdata/new_search_pv/" + date + "/"
									+ hour
									+ "/00/ttslave30.sds.cnz.alimama.com"),
							SequenceFileInputFormat.class, PvMap.class);
				}
			}
		// MultipleInputs.addInputPath(conf, pvPath,
		// SequenceFileInputFormat.class, PvMap.class);
		MultipleInputs.addInputPath(conf, clickUserAuctionPath,
				TextInputFormat.class, ClickUserAuctionMap.class);
		FileOutputFormat.setOutputPath(conf, pvClickPath);
		conf.setReducerClass(Reduce.class);
		JobClient.runJob(conf);
		return 0;
	}

	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(),
				new ClickUserAuctionInnerJoinPv(), args);
		System.exit(res);
	}

}
